package com.wutoon.etl.handler;

import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.wutoon.etl.aop.PowerJobLogReport;
import com.wutoon.etl.constant.CommonConstant;
import com.wutoon.etl.enums.AviatorResultStatusEnum;
import com.wutoon.etl.handler.function.DealResponseFunction;
import com.wutoon.etl.mapper.BizOperateRecordManage;
import com.wutoon.etl.pojo.AviatorResult;
import com.wutoon.etl.pojo.Http2DbTask;
import com.wutoon.etl.util.AviatorUtils;
import com.wutoon.etl.util.CommonUtil;
import com.wutoon.etl.util.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.http.*;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;

import java.lang.reflect.InvocationTargetException;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicMarkableReference;
import java.util.stream.Collectors;

/**
 * @author 武兴云/72176468
 * @version 1.0
 * @date 2024/1/22 17:48
 * @description http方式采集数据
 */
@Slf4j
public class HttpCollector implements Runnable{
    private final BlockingQueue<Object[]> queue;

    private final Http2DbTask task;

    private final CountDownLatch countDownLatch;

    private final AtomicMarkableReference<String> msg;

    private final List<String> subDim;

    private final RestTemplate restTemplate;

    private final BizOperateRecordManage bizOperateRecordManage;

    private final ThreadPoolExecutor workerPool;

    private RequestAttributes requestAttributes;

    public HttpCollector(ThreadPoolExecutor workerPool, RestTemplate restTemplate, BizOperateRecordManage bizOperateRecordManage,
                         BlockingQueue<Object[]> queue, Http2DbTask task, CountDownLatch countDownLatch,
                         AtomicMarkableReference<String> msg, List<String> subDim, RequestAttributes requestAttributes) {
        this.queue = queue;
        this.task = task;
        this.countDownLatch = countDownLatch;
        this.msg = msg;
        this.subDim = subDim;
        this.restTemplate = restTemplate;
        this.bizOperateRecordManage = bizOperateRecordManage;
        this.workerPool = workerPool;
        this.requestAttributes = requestAttributes;
    }

/**
 * 不同于jdbcReader和jdbcWriter，http查询要慢很多，同时要控制并发避免对下游造成攻击
 * 要保证http采集速度和避免任务倾斜，必要时开启二级线程池
 * 异常时及时感知，清空任务队列，避免忙跑
 *
 * @author 武兴云/72176468
 * @date 2024/2/29 9:05
 * @update_by 武兴云/72176468
 * @update_at 2024/2/29 9:05
 * @creed if you have doubt , please contact me !!!
 */
@Override
public void run() {
    log.info("http collector start ......");
    RequestContextHolder.setRequestAttributes(requestAttributes);
    PowerJobLogReport.report("http collector start ");
    long dataSize = 0;
    long start = System.currentTimeMillis();

    try {
        // 在此处并发调用http接口
        Http2DbTask.HttpCollectSourceConfig sourceConfig = task.getCollectSource();
        Iterator<String> iterator = subDim.iterator();
        // 有数据，并且主线程没有停止子线程
        while (iterator.hasNext() && !Thread.currentThread().isInterrupted()) {
            String dimension = iterator.next();

            // 组装请求头
            HttpHeaders headers = buildHeaders(sourceConfig);
            // 组装请求体
            JSONObject bodyTemplate = sourceConfig.getBodyTemplate() == null ? new JSONObject() : sourceConfig.getBodyTemplate();
            String template = bodyTemplate.toJSONString();

            // 打包变化参数位置
            Map<String, Object> urlParam = new HashMap<>();
            Map<String, Object> bodyParam = new HashMap<>();
            Http2DbTask.HttpCollectParamConfig page = packParamMap(sourceConfig, dimension, urlParam, bodyParam);

            // 请求远程接口
            AviatorResult<Integer> pageTotalResult = requestRemoteData(sourceConfig, headers, template, sourceConfig.getUrl(), urlParam, bodyParam, msg, null);
            // 发生异常退出
            if (msg.isMarked()) {
                return;
            }
            if (pageTotalResult.getResult() <= 1) {
                continue;
            }
            // 所有子线程是否结束的计数器，
            int countDownNum = pageTotalResult.getResult().equals(Integer.MAX_VALUE) ? CommonConstant.NUMBER_ZERO : pageTotalResult.getResult()-1;
            CountDownLatch subThreadEnd = new CountDownLatch(countDownNum);
            // 如果页数过多，是否开启二级并发调用
            for (int i = 1; i < pageTotalResult.getResult(); i++) {
                PowerJobLogReport.report("http collector submit %s page request task", i);
                // 默认分页从1开始
                packParam(page.getKey(), i + page.getStartIndex(), urlParam, bodyParam);
                // 存在分页总数并且达到二级并发的阈值，则并发处理
                if (pageTotalResult.getResult() > sourceConfig.getSecondConcurrency() && pageTotalResult.getResult() != Integer.MAX_VALUE) {
                    // 使用时动态修改核心大小
                    if (workerPool.getCorePoolSize() != CommonConstant.SECOND_POOL_CORE_SIZE) {
                        workerPool.setCorePoolSize(CommonConstant.SECOND_POOL_CORE_SIZE);
                    }
                    // 变化的参数要克隆一份，否则异步之后会被后面的循环覆盖值
                    Map<String, Object> clonedUrlParamMap = JSON.parseObject(JSON.toJSONString(urlParam), Map.class);
                    Map<String, Object> clonedBodyParamMap = JSON.parseObject(JSON.toJSONString(bodyParam), Map.class);
                    // 使用当前任务定义的线程池，使用公共线程池存在问题：
                    // 1、如果子线程存在异常，不能关闭线程池否则会对其他任务影响
                    // 2、如果子线程存在异常，已提交的大量任务队列实际没有执行的必要了，依然会占用大量资源持续执行，也会给下游服务带来压力
                    workerPool.submit(() -> requestRemoteData(sourceConfig, headers, template, sourceConfig.getUrl(), clonedUrlParamMap, clonedBodyParamMap, msg, subThreadEnd));
                    continue;
                }
                subThreadEnd.countDown();
                // 不开启二级并发时，是线性执行，要再做一次判断，是否存在异常的结果，避免占用资源做无用功，加速异常场景的任务结束
                if (msg.isMarked()) {
                    throw new RuntimeException(msg.getReference());
                }
                // 此处不pageTotalResult承接返回值，否则可能造成分页大小不一样，数据紊乱
                AviatorResult<Integer> currentPageResult = requestRemoteData(sourceConfig, headers, template, sourceConfig.getUrl(), urlParam, bodyParam, msg, subThreadEnd);
                if (currentPageResult.getResult() == -1) {
                    break;
                }
            }
            long subStart = System.currentTimeMillis();
            long now = System.currentTimeMillis();
            // 循环观测子线程是否存在异常，及时感知，否则发生异常时感知不及时，会等到全部结束或一小时才结束任务
            while ((now - subStart) / 1000 < CommonConstant.CONSUMER_TIME_OUT) {
                // 消费结束或存在异常，退出，否则继续观测
                if (subThreadEnd.await(CommonConstant.LOOP_OBSERVATION_TIME_OUT, TimeUnit.SECONDS) || msg.isMarked()) {
                    break;
                }
                now = System.currentTimeMillis();
                if ((now - subStart) / 1000 >= CommonConstant.CONSUMER_TIME_OUT) {
                    // 任务超时的场景
                    log.error("http collector task time out");
                    workerPool.shutdownNow();
                    throw new RuntimeException("task time out in " + CommonConstant.CONSUMER_TIME_OUT);
                }
            }

            if (!msg.isMarked()) {
                log.info("DataTransferHandler end success");
            } else {
                log.warn("DataTransferHandler not end in {} seconds, shutdown now", CommonConstant.CONSUMER_TIME_OUT);
                // 关闭当前正在执行的子线程，只会interrupt正在执行的线程，需要在子线程判断终止标志
                // 关闭线程池，清空任务列表，子线程可能不会被立刻中断取消，但是没关系正常执行完即可，随后所有子线程不再存活，主流程关闭，会被垃圾回收器回收
                // shutdown不会清空任务列表，而是等已提交的任务全部执行完，不适合此处,使用shutdownNow
                workerPool.shutdownNow();
                throw new RuntimeException(msg.getReference());
            }
        }
    } catch (Exception e) {
        log.error("http collector error:, ", e);
        String stackTraceString = CommonUtil.getStackTraceString(e);
        msg.set(stackTraceString, true);
    } finally {
        countDownLatch.countDown();
        log.info("http collector end duration:{}, dataSize:{}", System.currentTimeMillis() - start, dataSize);
        PowerJobLogReport.report("JdbcReader fetch duration:%s, dataSize:%s" , System.currentTimeMillis() - start, dataSize);
        RequestContextHolder.resetRequestAttributes();
    }
}

    /**
     * 本剧入参组装请求头
     *
     * @param sourceConfig Http2DbTHttpCollectSourceConfig
     * @return HttpHeaders
     * @author 武兴云/72176468
     * @date 2024/2/28 15:13
     * @update_by 武兴云/72176468
     * @update_at 2024/2/28 15:13
     * @creed if you have doubt , please contact me !!!
     */
    private static HttpHeaders buildHeaders(Http2DbTask.HttpCollectSourceConfig sourceConfig) {
        HttpHeaders headers = new HttpHeaders();
        JSONObject head = sourceConfig.getHead();
        head.forEach((key, value) -> headers.add(key, (String) value));
        if (!headers.containsKey(HttpHeaders.CONTENT_TYPE)) {
            headers.add(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE);
        }
        return headers;
    }

    /**
     * 打包参数map
     *
     * @param sourceConfig Http2DbTHttpCollectSourceConfig
     * @param dimension String
     * @param urlParam Map<String
     * @param bodyParam Object>
     * @return Http2DbTHttpCollectParamConfig
     * @author 武兴云/72176468
     * @date 2024/2/28 15:13
     * @update_by 武兴云/72176468
     * @update_at 2024/2/28 15:13
     * @creed if you have doubt , please contact me !!!
     */
    private Http2DbTask.HttpCollectParamConfig packParamMap(Http2DbTask.HttpCollectSourceConfig sourceConfig, String dimension, Map<String, Object> urlParam, Map<String, Object> bodyParam) {
        packParam(sourceConfig.getDimension().getKey(), dimension, urlParam, bodyParam);
        Http2DbTask.HttpCollectParamConfig page = sourceConfig.getPage();
        packParam(page.getKey(), page.getStartIndex(), urlParam, bodyParam);
        Http2DbTask.HttpCollectParamConfig beginTime = sourceConfig.getBeginTime();
        packParam(beginTime.getKey(), beginTime.getValue(), urlParam, bodyParam);
        Http2DbTask.HttpCollectParamConfig endTime = sourceConfig.getEndTime();
        packParam(endTime.getKey(), endTime.getValue(), urlParam, bodyParam);
        Http2DbTask.HttpCollectParamConfig pageSize = sourceConfig.getPageSize();
        packParam(pageSize.getKey(), pageSize.getSize(), urlParam, bodyParam);
        return page;
    }

    /**
     * 渲染数据，
     * 请求远程，
     * 表达式判断
     * 返回格式没有页数并且没有下一页的返回-1
     * 返回格式没有页数的，返回Integer.MAX_VALUE
     *
     * @param sourceConfig Http2DbTHttpCollectSourceConfig
     * @param headers      HttpHeaders
     * @param template     Template
     * @param url          String
     * @param urlParam     Map<String
     * @param bodyParam    Object>
     * @param msg
     * @param subThreadEnd
     * @return AviatorResult<Integer>
     * @author 武兴云/72176468
     * @date 2024/2/28 10:55
     * @update_by 武兴云/72176468
     * @update_at 2024/2/28 10:55
     * @creed if you have doubt , please contact me !!!
     */
    private AviatorResult<Integer> requestRemoteData(Http2DbTask.HttpCollectSourceConfig sourceConfig, HttpHeaders headers,
                                                     String template, String url, Map<String, Object> urlParam,
                                                     Map<String, Object> bodyParam, AtomicMarkableReference<String> msg,
                                                     CountDownLatch subThreadEnd) {
        try {
            // 替换参数模板
            HttpEntity<String> requestEntity;
            if (Objects.equals(HttpMethod.resolve(sourceConfig.getMethod()), HttpMethod.GET)) {
                requestEntity = new HttpEntity<>(null, headers);
            } else {
                String realBody = replacePlaceholders(template, bodyParam);
                requestEntity = new HttpEntity<>(realBody, headers);
            }
            long beforeSend = System.currentTimeMillis();
            // 发送请求
            ResponseEntity<JSONObject> responseEntity =
                    restTemplate.exchange(url, Objects.requireNonNull(HttpMethod.resolve(sourceConfig.getMethod())),
                            requestEntity, JSONObject.class, urlParam);
            // 判断是否成功
            JSONObject responseBody = responseEntity.getBody();
            long afterSend = System.currentTimeMillis();
            AviatorResult<Boolean> successResult = AviatorUtils.executeExpression(sourceConfig.getJudgeSuccessExpression(), responseBody, Boolean.class);
            if (successResult.getResult() == null || successResult.getResult().equals(Boolean.FALSE)) {
                throw new RuntimeException("collect data error");
            }
            // 遍历数据，生产数据
            List<Object> data = unStructResponse(sourceConfig, responseBody);
            String bizOperateRecordId = IdUtil.getSnowflakeNextIdStr();
            // 请求记录保存
            bizOperateRecordManage.saveBizOperateRecord(data, requestEntity.getBody(), bizOperateRecordId, DateUtil.generateCurrentDayId(DateUtil.YYYYMMDD), beforeSend, afterSend, sourceConfig.getSystem());
            List<String> fields = Arrays.stream(sourceConfig.getFieldList().split(",")).map(String::trim).collect(Collectors.toList());
            for (Object datum : data) {
                JSONObject item = JSON.parseObject(JSON.toJSONString(datum));
                Object[] row = new Object[fields.size()];
                for (int j = 0; j < fields.size(); j++) {
                    String columnName = fields.get(j);
                    // 所有http采集方式的ods表都要加一个请求批次号biz_operate_record_id，用于关联请求记录
                    Object columnValue = item.get(columnName);
                    if (StringUtils.equals(columnName, CommonConstant.BIZ_OPERATE_RECORD_ID)) {
                        columnValue = bizOperateRecordId;
                    }
                    row[j] = columnValue;
                }
                // 请求记录在贴源层数据表记录，用于关联请求记录，快速找到采集记录
                // 如果队列满了会阻塞等待一分钟
                queue.offer(row, CommonConstant.PRODUCER_TIMEOUT_SECONDS, TimeUnit.SECONDS);
            }
            // 观察该日志，看数据队列的实时容量，判断是生产者和消费者是否平衡。如果容量偏满则增加消费者
            log.info("Thread-name:{},producing data,current get data:{},queue size:{}",Thread.currentThread().getName(), data.size(), queue.size());
            // 返回分页
            // 没有返回总页数的场景下，判断返回不满足一页则是最后一页
            if (sourceConfig.getPageTotalExpression().equals(CommonConstant.NO_PAGETOTAL) && (CollectionUtils.isEmpty(data) || data.size() < sourceConfig.getPageSize().getSize())) {
                return AviatorResult.success(-1);
            }
            return AviatorUtils.executeExpression(sourceConfig.getPageTotalExpression(), responseBody, Integer.class);
        } catch (Exception e) {
            log.error("http collector request remote error,param:{}, exception:", sourceConfig, e);
            String stackTraceString = CommonUtil.getStackTraceString(e);
            msg.set(stackTraceString, true);
        } finally {
            if (subThreadEnd != null) {
                subThreadEnd.countDown();
            }
        }
        return new AviatorResult<>(AviatorResultStatusEnum.EXECUTE_ERROR, "", 0);

    }

    private List<Object> unStructResponse(Http2DbTask.HttpCollectSourceConfig sourceConfig, JSONObject responseBody) {
        if (sourceConfig.getDataExpression().startsWith(CommonConstant.REF_METHOD)) {
            String functionName = sourceConfig.getDataExpression().split(CommonConstant.REF_METHOD)[1];
            try {
                Class<?> function = Class.forName(functionName);
                DealResponseFunction instance = (DealResponseFunction) function.getDeclaredConstructor().newInstance();
                return instance.unStruct(responseBody);
            } catch (ClassNotFoundException | NoSuchMethodException | InstantiationException | IllegalAccessException |
                     InvocationTargetException e) {
                throw new RuntimeException(e);
            }
        } else {
            AviatorResult<ArrayList> dataResult = AviatorUtils.executeExpression(sourceConfig.getDataExpression(), responseBody, ArrayList.class);
            return dataResult.getResult();
        }
    }

    /**
     * 组装参数
     *
     * @param key String
     * @param param Object
     * @param urlParam Map<String
     * @param bodyParam Object>
     * @author 武兴云/72176468
     * @date 2024/2/28 15:14
     * @update_by 武兴云/72176468
     * @update_at 2024/2/28 15:14
     * @creed if you have doubt , please contact me !!!
     */
    private void packParam(String key, Object param, Map<String, Object> urlParam, Map<String, Object> bodyParam) {
        urlParam.put(key, param);
        bodyParam.put(key, param);
    }

    /**
     * 参数占位符替换，格式{xxx}
     *
     * @param template String
     * @param values Map<String
     * @return String
     * @author 武兴云/72176468
     * @date 2024/2/28 15:01
     * @update_by 武兴云/72176468
     * @update_at 2024/2/28 15:01
     * @creed if you have doubt , please contact me !!!
     */
    public String replacePlaceholders(String template, Map<String, Object> values) {
        for (Map.Entry<String, Object> entry : values.entrySet()) {
            template = template.replaceAll("\\{" + entry.getKey() + "}", String.valueOf(entry.getValue()));
        }
        return template;
    }

}
